package com.lzyx;

import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;
import reactor.netty.DisposableServer;
import reactor.netty.tcp.TcpServer;

import java.nio.charset.StandardCharsets;

@Slf4j
public class NettyStartApplication {

    public static void main(String[] args) {
        log.info("application start ------------");
        DisposableServer server = TcpServer
                .create()
                .host("localhost")
                .port(8080)
                .doOnConnection(connection -> {
                    log.info("on connection, remote address: {}", connection.channel().remoteAddress().toString());
                })
                .doOnChannelInit((connectionObserver, channel, remoteAddress) -> {
                    log.info("on channel init, remote address: {}|{}|{}", connectionObserver, channel, remoteAddress);
                    channel.pipeline()
                            .addFirst(new ChannelInitializer<SocketChannel>() {
                                @Override
                                protected void initChannel(SocketChannel ch) throws Exception {
                                    log.debug("init handle");
                                    ch.pipeline().remove(this);
                                    ch.pipeline()
                                            .addFirst("debug01", new ChannelInboundHandlerAdapter() {
                                                @Override
                                                public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                                    log.debug("channel is active: {}", ctx.pipeline());
                                                    super.channelActive(ctx);
                                                }

                                                @Override
                                                public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                                    log.debug("channel is read: {}", msg);
                                                    ByteBuf byteBuf = (ByteBuf) msg;
                                                    byte[] b = new byte[byteBuf.readableBytes()];
                                                    byteBuf.readBytes(b);
                                                    ReferenceCountUtil.release(byteBuf);
                                                    String result = new String(b, StandardCharsets.UTF_8);
                                                    super.channelRead(ctx, result + "qqq");
                                                }

                                                @Override
                                                public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
                                                    log.debug("channel is read complete:");
                                                    super.channelReadComplete(ctx);
                                                }
                                            });
//                                    ch.pipeline().addAfter("debug01", "debug02", new SimpleChannelInboundHandler<String>() {
//                                        @Override
//                                        protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
//                                            log.debug(msg);
//                                        }
//                                    });
                                    log.debug("finish init: {}", ch.pipeline().toMap());
                                }
                            });
                })
                .doOnBind(tcpServerConfig -> {
                    log.info("on bind, port: {}", tcpServerConfig.bindAddress());
                })
                .handle((nettyInbound, nettyOutbound) ->
                        nettyInbound
                                .receiveObject()
                                .cast(String.class)
                                .map(o -> {
                                    log.debug("in reactor handle: {}", o);
                                    return o;
                                })
                                .onErrorContinue((throwable, o) -> log.warn("handle error: {}", throwable.getMessage()))
                                .flatMap(s -> nettyOutbound.sendString(Mono.just("abc+"+s)).then())
                                .then()
                )
                .bindNow();

        server
                .onDispose()
                .block();
    }
}
